package qdu.edu.com.fushanf4.service.text.search;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import qdu.edu.com.fushanf4.utils.ConfigurationUtil;
import qdu.edu.com.fushanf4.utils.UtilConstants;

import java.io.IOException;
import java.util.*;

/**
 * KeyOut为   生成的一句话
 * ValueOut为 多个关键字
 */
public class TextGenerationByKeysMapper extends Mapper<LongWritable, Text, Text, Text> {
    private final HashMap<String, ArrayList<String>> hashMap = new HashMap<>(8196);
    private final List<String[]> keysInputs = new ArrayList<>(5);
    private final Text keyOut = new Text();
    private final Text valueOut = new Text();
    private boolean flag = false;
    private String splicing = "";

    @Override
    protected void setup(Context context) throws IOException {
        if (null != context.getConfiguration().get("keysInputPath")) {
            String keysInputPath = context.getConfiguration().get("keysInputPath");
            String fileName = "part-r-00000";
            List<String> keysSource = IOUtils.readLines(FileSystem.get(ConfigurationUtil.getConf()).open(new Path(keysInputPath, fileName)));

            for (String s : keysSource) {
                keysInputs.add(s.split(UtilConstants.RegexOfString.HORIZONTAL_TABULATION));
            }
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) {
        String line = value.toString();
        String[] split = line.split(UtilConstants.RegexOfString.HORIZONTAL_TABULATION);
        String[] two = split[0].split("");
        if (null != hashMap.get(two[0])) {
            hashMap.get(two[0]).add(two[1]);
        } else {
            hashMap.put(two[0], new ArrayList<>(Collections.singleton(two[1])));
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 遍历ArrayList，深度优先遍历，查找到一个就结束
        // 可能出现叠词，造成死循环，例如一一  已解决，如果拼接的字符串中已经含有该关键词，就结束本次方法调用
        String temp;
        for (String[] keys : keysInputs) {
            splicing = "";                                  // 循环开始时初始化
            temp = splicing;
            splicing = temp + keys[0];
            for (int i=0; i<keys.length-1; i++) {
                searchValues(keys[i], keys[i + 1]);
            }
            if (!splicing.contains(keys[keys.length-1])) {  // 防止叠词，以及另一种情况   一二三一  死循环
                temp = splicing;
                splicing = temp + keys[keys.length-1];
            }
            keyOut.set(splicing);
            valueOut.set(Arrays.toString(keys));
            context.write(keyOut, valueOut);
        }
    }

    /**
     * 该方法用来根据key查找value，然后将value作为新的可以继续查找value
     * @param key 作为关键字
     */
    private void searchValues(String key, String nextKey) {
        String temp = splicing;
        ArrayList<String> values = hashMap.get(key);
        if (values == null) {                   // 已经最深，不能继续递归了，切，然后return
            splicing = splicing.substring(0, splicing.length()-1);
            return;
        }
        if (splicing.contains(nextKey)) { // 判断该键对应的所有值中是否已经含有了需要的nextKey
            flag = true;
            return;
        }
        for (String valueTemp : values) {
            if (nextKey.equals(valueTemp)) {
                temp = splicing;
                splicing = temp + valueTemp;
                flag = true;
                return;
            }
        }
        for (String valueTemp : values) {
            if (splicing.contains(valueTemp)) { // 判断是否已经包含了汉字
                continue;
            }
            splicing = temp + valueTemp;
            temp = splicing;
            if (valueTemp.equals(nextKey)) {    // 如果找到下一个键，就结束本次循环
                flag = true;
                return;      // 找到了，处理
            }
            searchValues(valueTemp, nextKey);
            if (!flag) {
                splicing = splicing.substring(0, splicing.length()-1);
            } else {
                return;
            }
        }
    }
}
